package com.allen.flink.batch.join;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;

/**
 * 功能:右外连接
 *
 * @date: 2020-03-24 13:11
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class OuterRightJoin {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //第一个数据集Tuple 用户Id 用户名
        ArrayList<Tuple2<Integer, String>> list1 = new ArrayList<>();
        list1.add(new Tuple2<>(1, "Mars"));
        list1.add(new Tuple2<>(2, "Tom"));
        list1.add(new Tuple2<>(4, "Alex"));
        
        //第二个数据集Tuple 用户Id 用户城市
        ArrayList<Tuple2<Integer, String>> list2 = new ArrayList<>();
        list2.add(new Tuple2<>(1, "BeiJing"));
        list2.add(new Tuple2<>(2, "ShangHai"));
        list2.add(new Tuple2<>(3, "ChengDu"));
        
        DataSet<Tuple2<Integer, String>> ds1 = env.fromCollection(list1);
        DataSet<Tuple2<Integer, String>> ds2 = env.fromCollection(list2);
        
        /**
         * 右外连接
         * list1 tuple中元素可能为null
         */
        ds1.rightOuterJoin(ds2).where(0).equalTo(0).with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String, String>>() {
            @Override
            public Tuple3<Integer, String, String> join(final Tuple2<Integer, String> value1, final Tuple2<Integer, String> value2) throws Exception {
                if(value1==null){
                    return new Tuple3<>(value2.f0, null, value2.f1);
                }
                return new Tuple3<>(value1.f0, value1.f1, value2.f1);
            }
        }).print();
        
        
    }
}
